package com.pai4j.ugc.messaegqueue.service;
import com.pai4j.common.bean.PAIPageResponseBeanUtil;
import com.pai4j.common.bean.PAIResponseBean;
import com.pai4j.common.bean.PAIResponseBeanUtil;
import com.pai4j.common.enums.MessageTypeEnum;
import com.pai4j.common.enums.RedisKeyEnum;
import com.pai4j.common.enums.RoleCodeEnum;
import com.pai4j.common.enums.messagequeue.MessageQueueEnum;
import com.pai4j.common.service.messagequeue.producer.MessageQueueProducer;
import com.pai4j.common.util.DateUtils;
import com.pai4j.common.util.HtmlUtil;
import com.pai4j.common.util.JsonUtil;
import com.pai4j.common.util.RedisUtil;
import com.pai4j.domain.vo.request.mess.ChatMessageRequestVO;
import com.pai4j.domain.vo.request.mess.MessageBaseVO;
import com.pai4j.domain.vo.response.UserBaseResponseInfoVO;
import com.pai4j.domain.vo.response.mess.ChatMessageResponseVO;
import com.pai4j.domain.vo.response.mess.SubjectAccountArticleMessageVO;
import com.pai4j.domain.vo.response.subjectaccount.ArticleBaseResponseVO;
import com.pai4j.remote.pgc.SubjectArticleClient;
import com.pai4j.ugc.repository.dao.IChatMessageDAO;
import com.pai4j.ugc.repository.dao.IInteractionMessageDAO;
import com.pai4j.ugc.repository.entity.ChatMessageEntity;
import com.pai4j.ugc.repository.entity.InteractionMessageEntity;
import com.pai4j.ugc.service.AccountService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class AbstractPersistentMessageService {


    @Autowired
    private IChatMessageDAO chatMessageDAO;

    @Resource
    private IInteractionMessageDAO interactionMessageDAO;

    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private AccountService accountService;

    @Autowired
    private MessageQueueProducer<Object, Long> messageQueueProducer;

    @Autowired
    private SubjectArticleClient subjectArticleClient;

    /**
     * 《离线场景》持久化消息队列，保存离线消息，同时会将数据同步到ES用于后续的RAG问答和消息搜索
     *
     * @param messVo
     * @return
     */
    public boolean persistent(MessageBaseVO messVo) {
        log.info("消息持久化开始，senderId: {}, receiverId: {}, sessionId: {}, type: {}", 
                messVo.getSenderId(), messVo.getReceiverId(), messVo.getSessionId(), messVo.getType());
        Integer messageType = messVo.getType();
        // 聊天消息、文件消息、推荐卡片消息统一处理（都是聊天记录中的消息）
        if(messageType.equals(MessageTypeEnum.CHAT_TEXT.getType()) 
                || messageType.equals(MessageTypeEnum.CHAT_FILE.getType())
                || messageType.equals(MessageTypeEnum.RECOMMEND_USER_CARD.getType())
                || messageType.equals(MessageTypeEnum.RECOMMEND_GROUP_CARD.getType())){
            // 生成消息唯一标识（基础部分）
            String baseMessageUuid = generateMessageUuid(messVo);
            
            ChatMessageEntity mess = convertChatMesssage(messVo);
            // 取消息接收人信息，判断是否群聊场景
            String receiverId = messVo.getReceiverId();
            UserBaseResponseInfoVO receiver = accountService.accountInfo(receiverId);
            boolean groupMessage = RoleCodeEnum.GROUP.getCode().equals(receiver.getRoleCode());
            List<String> receivers = new ArrayList<>();
            if (groupMessage) {
                // 角色翻转，已群组作为发送人
                mess.setSender(receiverId);
                // 设置代理消息发送用户
                mess.setProxySender(messVo.getSenderId());
                // 查询实际接收人，群组下所有用户
                Set<String> groupUserIds = accountService.getGroupUserIds(receiverId);
                receivers.addAll(groupUserIds);
            } else {
                receivers.add(receiverId);
            }
            List<String> persistenceReceivers = new ArrayList<>(receivers);
            persistenceReceivers.add(messVo.getSenderId());
            if (!receivers.isEmpty()) {
                String firstReceiverUserId = receivers.get(0);
                String firstReceiverMessageUuid = baseMessageUuid + ":" + firstReceiverUserId;
            
                if (chatMessageDAO.existsByMessageUuid(firstReceiverMessageUuid)) {
                    return true;
                }
            } else {
                // 没有接收者（发送人是唯一群成员），直接返回成功
                log.warn("群聊消息没有接收者（发送人是唯一成员），跳过。groupId: {}, senderId: {}",
                        receiverId, messVo.getSenderId());
                return true;
            }
            for (String receiverUserId : persistenceReceivers) {
                
                // 为每个接收者创建一份消息副本（群聊场景需要为每个成员创建独立记录）
                ChatMessageEntity messCopy = new ChatMessageEntity();
                BeanUtils.copyProperties(mess, messCopy);
                // 确保是新记录（清除可能的ID）
                messCopy.setId(null);
                // 【关键】设置当前接收者ID
                messCopy.setReceiver(receiverUserId);
                // 如果是发送者本人，标记为已读，避免未读计数
                if (receiverUserId.equals(messVo.getSenderId())) {
                    messCopy.setIsRead(true);
                }
                
                String receiverMessageUuid;
                if (groupMessage) {
                    // 群聊：baseUuid包含groupId，需要拼接具体接收者ID以区分不同成员的消息记录
                    receiverMessageUuid = baseMessageUuid + ":" + receiverUserId;
                } else {
                    // 单聊：baseUuid已包含receiverId，无需再拼接（避免冗余）
                    receiverMessageUuid = baseMessageUuid;
                }
                messCopy.setMessageUuid(receiverMessageUuid);
            
                if (chatMessageDAO.existsByMessageUuid(receiverMessageUuid)) {
                    log.warn("接收者 {} 的消息已存在，跳过。messageUuid: {}", receiverUserId, receiverMessageUuid);
                    continue;
                }
                
                try {
                    // 执行数据库插入
                    messCopy = chatMessageDAO.save(messCopy);
                    log.info("消息持久化成功，messId: {}, messageUuid: {}, receiver: {}", 
                            messCopy.getId(), messCopy.getMessageUuid(), receiverUserId);
                } catch (Exception e) {
                    log.error("消息持久化失败。receiver: {}, messageUuid: {}, error: {}", 
                            receiverUserId, receiverMessageUuid, e.getMessage(), e);
                    // 如果是唯一索引冲突，说明消息已存在，跳过即可
                    // 其他异常继续处理下一个接收者
                    continue;
                }
                
                // 加入聊天缓存
                if (groupMessage) {
                    // 发送者本人：使用"发送者自己的那条DB记录ID"写入发送者会话缓存，确保回填一致
                    if (receiverUserId.equals(messVo.getSenderId())) {
                        addUserMessCache(messCopy.getProxySender(), messVo.getReceiverId(), messCopy.getId(), messCopy.getId());
                    } else {
                        // 非发送者：为当前接收者添加群会话消息
                        addUserMessCache(receiverUserId, messVo.getReceiverId(), messCopy.getId(), messCopy.getId());
                    }
                } else {
                    // 一对一：保持原有双向写入
                    this.addUserMessCache(messCopy);
                }
                
                // 加入聊天列表
                if (groupMessage) {
                    // 发送者与群的对话列表刷新（多次调用无副作用，ZSet会被最新时间覆盖）
                    this.addOrRefreshMessListCache(messCopy.getProxySender(), messVo.getReceiverId());
                    // 接收者与群的对话列表刷新
                    this.addOrRefreshMessListCache(receiverUserId, messVo.getReceiverId());
                    // 【关键】群聊未读消息+1：跳过发送者本人
                    if (!receiverUserId.equals(messVo.getSenderId())) {
                        addUnreadMessCountCache(receiverUserId, messCopy.getSender());
                        log.debug("群聊消息未读计数已更新，receiver: {}, sender: {}", receiverUserId, messCopy.getSender());
                    }
                } else {
                    this.addOrRefreshMessListCache(messVo.getSenderId(), receiverUserId);
                    this.addOrRefreshMessListCache(receiverUserId, messVo.getSenderId());
                    // 【关键】未读消息+1（只要持久化就增加未读计数，确保离线消息有红点）
                    addUnreadMessCountCache(receiverUserId, messVo.getSenderId());
                    log.debug("一对一消息未读计数已更新，receiver: {}, sender: {}", receiverUserId, messVo.getSenderId());
                }
                /**
                 * 刷新用户对话列表
                 *
                 */
                this.refreshChattingCache(messVo);
                /**
                 * 发送新消息监听，新消息同步es
                 */
                messageQueueProducer.send(MessageQueueEnum.QUEUE_MESSAGE_SYNC_ES, messCopy);
            }
            log.info("消息持久化完成，总共处理 {} 个接收者", receivers.size());
        }
        if(messageType.equals(MessageTypeEnum.INTERACTION.getType())){
            // 生成消息唯一标识
            String messageUuid = generateInteractionMessageUuid(messVo);
            
            // 【幂等性检查】检查消息是否已存在，防止重复存储
            if (interactionMessageDAO.existsByMessageUuid(messageUuid)) {
                log.warn("互动消息已存在，跳过重复存储。messageUuid: {}, senderId: {}, receiverId: {}", 
                        messageUuid, messVo.getSenderId(), messVo.getReceiverId());
                return true;
            }
            
            InteractionMessageEntity interactionMessage = this.convertInteractionMessage(messVo);
            interactionMessage.setMessageUuid(messageUuid);
            
            // 【调试】输出转换后的实体字段
            log.info("【调试-转换后】messageType: {}, content: {}, commentId: {}, resourceId: {}, resourceType: {}, resourceTitle: {}", 
                    interactionMessage.getMessageType(), interactionMessage.getContent(), 
                    interactionMessage.getCommentId(), interactionMessage.getResourceId(), 
                    interactionMessage.getResourceType(), interactionMessage.getResourceTitle());
            
            interactionMessageDAO.save(interactionMessage);
            log.info("互动消息持久化成功，messageUuid: {}, senderId: {}, receiverId: {}", 
                    messageUuid, messVo.getSenderId(), messVo.getReceiverId());
        }
        return true;
    }


    private void refreshChattingCache(MessageBaseVO chatMessage) {
        // 刷新聊天对话列表
        Long msgTime = System.currentTimeMillis();
        ChatMessageRequestVO chattingRequest = new ChatMessageRequestVO();
        chattingRequest.setSenderId(chatMessage.getSenderId());
        chattingRequest.setReceiverId(chatMessage.getReceiverId());
        chattingRequest.setTime(msgTime);
        messageQueueProducer.send(MessageQueueEnum.QUEUE_CHATTING_LIST_REFRESH, chattingRequest);
        UserBaseResponseInfoVO accountInfo = accountService.accountInfo(chatMessage.getReceiverId());
        if (RoleCodeEnum.isUserRole(accountInfo.getRoleCode())) {
            // 如果消息接受账号是用户类型，也需要刷新用户账号列表
            chattingRequest.setSenderId(chatMessage.getReceiverId());
            chattingRequest.setReceiverId(chatMessage.getSenderId());
            messageQueueProducer.send(MessageQueueEnum.QUEUE_CHATTING_LIST_REFRESH, chattingRequest);
        } else if (RoleCodeEnum.GROUP.getCode().equals(accountInfo.getRoleCode())) {
            // 接受人是群组，需要刷新群组下所有用户对话列表
            Set<String> groupUsers = accountService.getGroupUserIds(chatMessage.getReceiverId());
            if (CollectionUtils.isNotEmpty(groupUsers)) {
                for (String groupUserId : groupUsers) {
                    if (groupUserId.equals(chatMessage.getSenderId())) {
                        continue;
                    }
                    chattingRequest.setSenderId(groupUserId);
                    chattingRequest.setReceiverId(chatMessage.getReceiverId());
                    messageQueueProducer.send(MessageQueueEnum.QUEUE_CHATTING_LIST_REFRESH, chattingRequest);
                }
            }
        }
    }

    /**
     * 刷新未读消息数缓存，用于新消息红点🔴通知
     *
     * @param currUserId
     * @param chatUserId
     * @return
     */
    private Long addUnreadMessCountCache(String currUserId, String chatUserId) {
        String unreadUserCacheCountKey = RedisKeyEnum.UN_READ_MESS_USER_SET_CACHE.getKey(currUserId);
        redisUtil.sadd(unreadUserCacheCountKey, chatUserId);
        String unreadCacheCountKey = RedisKeyEnum.UN_READ_MESS_COUNT_CACHE.getKey(currUserId, chatUserId);
        return redisUtil.increx(unreadCacheCountKey, RedisKeyEnum.UN_READ_MESS_COUNT_CACHE.getExpireTime());
    }

    private void clearUnreadMessCountCache(String currUserId, String chatUserId) {
        String unreadCacheCountKey = RedisKeyEnum.UN_READ_MESS_COUNT_CACHE.getKey(currUserId, chatUserId);
        redisUtil.set(unreadCacheCountKey, "0", RedisKeyEnum.UN_READ_MESS_COUNT_CACHE.getExpireTime());
        String unreadUserCacheCountKey = RedisKeyEnum.UN_READ_MESS_USER_SET_CACHE.getKey(currUserId);
        redisUtil.sremove(unreadUserCacheCountKey, chatUserId);
    }

    public Long getUnreadMessUserCountFromCache(String currUserId) {
        String unreadUserCacheCountKey = RedisKeyEnum.UN_READ_MESS_USER_SET_CACHE.getKey(currUserId);
        return redisUtil.ssize(unreadUserCacheCountKey);
    }

    public Set<String> getUnreadMessUserSetFromCache(String currUserId) {
        String unreadUserCacheCountKey = RedisKeyEnum.UN_READ_MESS_USER_SET_CACHE.getKey(currUserId);
        return redisUtil.smembers(unreadUserCacheCountKey);
    }


    /**
     * 查询两个用户指定时间段内的聊天消息
     *
     * @param sourceUserId
     * @param targetUserId
     * @param startTime
     * @param endTime
     * @param count
     * @return
     */
    public List<String> query(String sourceUserId, String targetUserId,
                              Long startTime, Long endTime, Integer count) {
        /**
         * 如果没有指定时间范围，默认查询最近7天的聊天数据
         */
        Date now = new Date();
        Date startDate = startTime == null ? DateUtils.getStartOfDay(new Date(now.getTime() - 6 * 24 * 60 * 60 * 1000L)) : new Date(startTime);
        Date endDate = endTime == null ? DateUtils.getEndOfDay(now) : new Date(endTime);
        Pageable pageable = PageRequest.of(0, count, Sort.Direction.DESC, "id");
        Page<ChatMessageEntity> chatMessageEntities =
                chatMessageDAO.findAllBySenderAndReceiverAndSendDateBefore(
                        sourceUserId, targetUserId, startDate, endDate, pageable);

        List<ChatMessageEntity> chatMessageEntityList = chatMessageEntities.getContent();
        if (CollectionUtils.isEmpty(chatMessageEntityList)) {
            return Collections.emptyList();
        }

        return chatMessageEntityList.stream().map(ChatMessageEntity::getMessage).collect(Collectors.toList());
    }

    /***
     * 查询两个人的聊天记录
     * 支持缓存失效时从数据库加载数据并回填缓存
     * 【重要修改】查询消息列表不再自动清除未读计数，需要前端主动调用markMessagesAsRead接口
     * 
     * @param currUserId 当前用户ID
     * @param chatUserId 聊天对象ID
     * @param lastId 最后一条消息ID（用于分页加载）
     * @param fixedMessageId 固定消息ID（用于消息定位）
     * @param size 查询数量
     * @return 聊天消息列表
     */
    public List<ChatMessageResponseVO>  getChatMessListFromCache(String currUserId, String chatUserId, Long lastId,
                                                                Long fixedMessageId, int size) {

       
        log.debug("用户 {} 查询与 {} 的聊天记录", currUserId, chatUserId);

        lastId = lastId == null ? Long.MAX_VALUE : lastId;
        String cacheKey = getUserMessRedisKey(currUserId, chatUserId);
        Set<String> cacheSet = redisUtil.zreverseRangeByScore(cacheKey, lastId, 0, size);
        
        // 如果缓存为空或数据不足，从数据库加载
        if (CollectionUtils.isEmpty(cacheSet) || cacheSet.size() < size) {
            log.info("缓存数据不足，从数据库加载聊天记录。当前用户：{}，聊天对象：{}，缓存数量：{}，期望数量：{}", 
                     currUserId, chatUserId, cacheSet == null ? 0 : cacheSet.size(), size);
            
            // 从数据库查询消息
            List<ChatMessageEntity> messagesFromDb = loadMessagesFromDatabase(currUserId, chatUserId, lastId, size);
            
            if (CollectionUtils.isNotEmpty(messagesFromDb)) {
                log.info("从数据库查询到 {} 条消息，开始回填缓存", messagesFromDb.size());
                
                // 【关键修复】回填缓存逻辑：只向当前查询的会话缓存写入，避免群聊重复
                for (ChatMessageEntity message : messagesFromDb) {
                    // 1. 刷新消息详情缓存（所有场景通用）
                    refreshMessCache(message);
                    
                    // 2. 只向当前查询的会话缓存键写入消息ID
                    // 这样可以避免群聊场景下的重复问题
                    // currUserId + chatUserId 就是当前用户查看的会话
                    addUserMessCache(currUserId, chatUserId, message.getId(), message.getId());
                    
                    log.debug("回填消息到缓存：messId={}, currUserId={}, chatUserId={}", 
                            message.getId(), currUserId, chatUserId);
                }
                
                // 重新从缓存查询
                cacheSet = redisUtil.zreverseRangeByScore(cacheKey, lastId, 0, size);
                log.info("缓存回填完成，重新查询到 {} 条消息", cacheSet == null ? 0 : cacheSet.size());
            } else {
                log.info("数据库中未查询到消息记录");
            }
        }
        
        if (CollectionUtils.isEmpty(cacheSet)) {
            return Collections.emptyList();
        }
        
        if (fixedMessageId != null) {
            cacheSet.remove(fixedMessageId.toString());
        }
        
        // 判断是否为群聊会话，用于后续过滤与清理
        boolean isGroupChat = RoleCodeEnum.GROUP.getCode().equals(accountService.accountInfo(chatUserId).getRoleCode());

        List<ChatMessageResponseVO> chatMessageResponseVOList = cacheSet.stream().map(cache -> {
            Long messId = Long.valueOf(cache);
            ChatMessageResponseVO messageResponse = getChatMessDetailFromCache(messId);
            messageResponse.setReceiver(accountService.accountInfo(messageResponse.getReceiverId()));
            messageResponse.setSender(accountService.accountInfo(messageResponse.getSenderId()));
            if (StringUtils.isNotBlank(messageResponse.getProxySenderId())) {
                messageResponse.setProxySender(accountService.accountInfo(messageResponse.getProxySenderId()));
            }
            // 安全校验：群聊场景下，如果缓存中混入了其他接收者的记录，主动清理并跳过
            if (isGroupChat && !StringUtils.equals(messageResponse.getReceiverId(), currUserId)) {
                // 从当前会话ZSET中移除这条错误的消息ID
                String wrongCacheKey = getUserMessRedisKey(currUserId, chatUserId);
                redisUtil.zremove(wrongCacheKey, String.valueOf(messId));
                return null;
            }
            return messageResponse;
        }).filter(Objects::nonNull)
          .sorted(Comparator.comparing(ChatMessageResponseVO::getMessId))
          .collect(Collectors.toList());

        if (fixedMessageId != null) {
            // 处理消息定位，简单处理，默认插入到消息尾部
            ChatMessageResponseVO fixedMessage = getChatMessDetailFromCache(fixedMessageId);
            fixedMessage.setReceiver(accountService.accountInfo(fixedMessage.getReceiverId()));
            fixedMessage.setSender(accountService.accountInfo(fixedMessage.getSenderId()));
            if (StringUtils.isNotBlank(fixedMessage.getProxySenderId())) {
                fixedMessage.setProxySender(accountService.accountInfo(fixedMessage.getProxySenderId()));
            }
            chatMessageResponseVOList.add(fixedMessage);
        }

        return chatMessageResponseVOList;
    }
    
    /**
     * 从数据库加载两个用户之间的聊天记录
     * 
     * @param currUserId 当前用户ID
     * @param chatUserId 聊天对象ID
     * @param lastId 最后一条消息ID
     * @param size 查询数量
     * @return 消息列表
     */
    private List<ChatMessageEntity> loadMessagesFromDatabase(String currUserId, String chatUserId, Long lastId, int size) {
        try {
            Pageable pageable = PageRequest.of(0, size);
            return chatMessageDAO.findChatMessagesByIdRange(currUserId, chatUserId, lastId, pageable);
        } catch (Exception e) {
            log.error("从数据库加载聊天记录失败。当前用户：{}，聊天对象：{}，错误：{}", currUserId, chatUserId, e.getMessage(), e);
            return Collections.emptyList();
        }
    }
    public Map<String, ChatMessageResponseVO> batchGetUserLastMess(String currUserId, Set<String> userIds) {
        Map<String, ChatMessageResponseVO> map = new HashMap<>();
        for (String userId : userIds) {
            String cacheKey = getUserMessRedisKey(currUserId, userId);
            Set<String> cacheSet =
                    redisUtil.zreverseRangeByScore(cacheKey, Long.MAX_VALUE, 0, 1);
            if (CollectionUtils.isNotEmpty(cacheSet)) {
                Long lastMessId = Long.valueOf(new ArrayList<>(cacheSet).get(0));
                ChatMessageResponseVO chatMessageResponseVO = getChatMessDetailFromCache(lastMessId);
                if (chatMessageResponseVO != null) {
                    map.put(userId, chatMessageResponseVO);
                }
            }
        }
        return map;
    }

    private void addOrRefreshMessListCache(String currUserId, String chatUserId) {
        String messUserKey = RedisKeyEnum.MESS_USER_LIST_KEY.getKey(currUserId);
        redisUtil.zadd(messUserKey, chatUserId, DateUtils.getCurrentTimeMillis());
    }

    private void addUserMessCache(ChatMessageEntity mess) {
        long messId = mess.getId();
        /***
         * 刷新消息详情缓存
         */
        refreshMessCache(mess);
        /***
         * 加入用户消息列表
         */
        addUserMessCache(
                mess.getProxySender() != null ? mess.getProxySender() : mess.getSender(),
                mess.getProxySender() != null ? mess.getSender() : mess.getReceiver(),
                messId, messId);
        addUserMessCache(mess.getReceiver(), mess.getSender(), messId, messId);
    }

    private void addUserMessCache(String sender, String receiver, Long messId, Long score) {
        String cacheKey = getUserMessRedisKey(sender, receiver);
        redisUtil.zadd(cacheKey, messId.toString(), score);
    }

    /***
     * 刷新消息缓存
     * @param mess
     */
    private void refreshMessCache(ChatMessageEntity mess) {
        String messKey = RedisKeyEnum.MESS_DETAIL_CACHE_KEY.getKey();
        redisUtil.hset(messKey, String.valueOf(mess.getId()), JsonUtil.toJsonString(mess),
                RedisKeyEnum.MESS_DETAIL_CACHE_KEY.getExpireTime());
    }

   private ChatMessageResponseVO getChatMessDetailFromCache(Long messId) {
    String messKey = RedisKeyEnum.MESS_DETAIL_CACHE_KEY.getKey();
    String messCache = redisUtil.hget(messKey, String.valueOf(messId));

    ChatMessageEntity userMessEntity = null;

    if (StringUtils.isBlank(messCache)) {
        // 缓存为空，从数据库查询
        Optional<ChatMessageEntity> optionalEntity = chatMessageDAO.findById(messId);
        if (optionalEntity.isPresent()) {
            userMessEntity = optionalEntity.get();
            // 更新缓存
            redisUtil.hset(messKey, String.valueOf(messId), JsonUtil.toJsonString(userMessEntity),
                    RedisKeyEnum.MESS_DETAIL_CACHE_KEY.getExpireTime());
        } else {
            return null;
        }
    } else {
        userMessEntity = JsonUtil.fromJson(messCache, ChatMessageEntity.class);
    }

    ChatMessageResponseVO messageResponse = new ChatMessageResponseVO();
    messageResponse.setMessId(userMessEntity.getId());
    messageResponse.setMessage(userMessEntity.getMessage());
    messageResponse.setImage(userMessEntity.getImage());
    messageResponse.setSenderId(userMessEntity.getSender());
    messageResponse.setProxySenderId(userMessEntity.getProxySender());
    messageResponse.setReceiverId(userMessEntity.getReceiver());
    messageResponse.setTime(userMessEntity.getSendDate().getTime());
    messageResponse.setIsRead(userMessEntity.getIsRead());
    messageResponse.setType(userMessEntity.getType());

    if (MessageTypeEnum.SUBJECT_ACCOUNT_ARTICLE.getType().equals(userMessEntity.getType())) {
        try {
            messageResponse.setSubjectAccountArticle(getArticleMessageVO(Long.valueOf(userMessEntity.getMessage())));
        } catch (NumberFormatException e) {
            log.warn("Failed to parse article ID from message: {}", userMessEntity.getMessage());
        }
    }

    messageResponse.setGroupMessage(StringUtils.isNotBlank(userMessEntity.getProxySender()));
    return messageResponse;
}



    private SubjectAccountArticleMessageVO getArticleMessageVO(Long articleId) {
        PAIResponseBean<ArticleBaseResponseVO> responseBean = subjectArticleClient.detail(articleId, false);
        if (!PAIResponseBeanUtil.isOk(responseBean)) {
            return null;
        }
        ArticleBaseResponseVO articleBaseResponseVO = responseBean.getData();
        if (articleBaseResponseVO == null) {
            return null;
        }
        SubjectAccountArticleMessageVO articleMessageVO = new SubjectAccountArticleMessageVO();
        BeanUtils.copyProperties(articleBaseResponseVO, articleMessageVO);
        articleMessageVO.setArticleId(articleBaseResponseVO.getId());
        return articleMessageVO;
    }

    public String getUserMessRedisKey(String sender, String receiver) {
        return RedisKeyEnum.USER_CHAT_MESS_CACHE_KEY.getKey(sender, receiver);
    }

    public PAIPageResponseBeanUtil<List<ChatMessageResponseVO>> pageMessage(String mess, int pageNo, int pageSize) {
        Pageable pageable = PageRequest.of(pageNo - 1, pageSize, Sort.by(Sort.Order.desc("id")));
        Page<ChatMessageEntity> chatMessEntityPage;
        if (StringUtils.isBlank(mess)) {
            chatMessEntityPage = chatMessageDAO.findAll(pageable);
        } else {
            chatMessEntityPage = chatMessageDAO.findAllByMessageLike("%"+mess+"%", pageable);
        }
        List<ChatMessageResponseVO> chatMessageResponseVOList = convertChatMessageResponseList(chatMessEntityPage.getContent());
        return PAIPageResponseBeanUtil.success(pageNo, pageSize, chatMessEntityPage.getTotalElements(), chatMessageResponseVOList);
    }

    private List<ChatMessageResponseVO> convertChatMessageResponseList(List<ChatMessageEntity> chatMessEntities) {
        if (CollectionUtils.isEmpty(chatMessEntities)) {
            return Collections.emptyList();
        }
        /**
         * 批量查询用户信息
         */
        Set<String> senderUserIds = chatMessEntities.stream().map(ChatMessageEntity::getSender).collect(Collectors.toSet());
        Set<String> proxySenderUserIds = chatMessEntities.stream().map(ChatMessageEntity::getProxySender).filter(Objects::nonNull).collect(Collectors.toSet());
        Set<String> receiverUserIds = chatMessEntities.stream().map(ChatMessageEntity::getReceiver).collect(Collectors.toSet());
        senderUserIds.addAll(receiverUserIds);
        if (CollectionUtils.isNotEmpty(proxySenderUserIds)) {
            senderUserIds.addAll(proxySenderUserIds);
        }
        Map<String, UserBaseResponseInfoVO> userMap = accountService.batchGet(senderUserIds);
        return chatMessEntities.stream().map(chat -> {
            ChatMessageResponseVO chatMessageResponse = new ChatMessageResponseVO();
            chatMessageResponse.setTime(chat.getSendDate().getTime());
            chatMessageResponse.setSender(userMap.get(chat.getSender()));
            if (StringUtils.isNotBlank(chat.getProxySender())) {
                chatMessageResponse.setProxySender(userMap.get(chat.getProxySender()));
            }
            chatMessageResponse.setReceiver(userMap.get(chat.getReceiver()));
            chatMessageResponse.setMessage(chat.getMessage());
            chatMessageResponse.setType(chat.getType());
            chatMessageResponse.setImage(chat.getImage());
            chatMessageResponse.setIsRead(chat.getIsRead());
            chatMessageResponse.setMessId(chat.getId());
            
            // 反序列化消息扩展数据（如推荐卡片数据）
            if (StringUtils.isNotBlank(chat.getMessageExt())) {
                try {
                    // 根据消息类型反序列化为对应的VO
                    // 注意：这里使用Object类型，前端需要根据type字段判断具体类型
                    Object messageExtObj = JsonUtil.fromJson(chat.getMessageExt(), Object.class);
                    chatMessageResponse.setMessageExt(messageExtObj);
                    log.debug("消息扩展数据反序列化成功，messId: {}, type: {}", chat.getId(), chat.getType());
                } catch (Exception e) {
                    log.error("消息扩展数据反序列化失败，messId: {}, type: {}", chat.getId(), chat.getType(), e);
                }
            }
            
            return chatMessageResponse;
        }).collect(Collectors.toList());
    }

    private ChatMessageEntity convertChatMesssage(MessageBaseVO messVo) {
        ChatMessageEntity mess = new ChatMessageEntity();
        mess.setSender(messVo.getSenderId());
        // 【关键修复】不在这里设置receiver，因为receiver需要在循环中为每个接收者单独设置
        // mess.setReceiver(messVo.getReceiverId()); // 移除这行，避免与循环中的设置冲突
        mess.setMessage(this.handleSpecialHtmlTag(HtmlUtil.xssEscape(messVo.getMessage())));
        mess.setSendDate(new Date());
        mess.setIsRead(false);
        mess.setImage(messVo.getImage());
        mess.setType(messVo.getType());
        
        // 序列化消息扩展数据（如推荐卡片数据）
        if (messVo.getMessageExt() != null) {
            try {
                String messageExtJson = JsonUtil.toJsonString(messVo.getMessageExt());
                mess.setMessageExt(messageExtJson);
                log.info("消息扩展数据序列化成功，type: {}, messageExt: {}", messVo.getType(), messageExtJson);
            } catch (Exception e) {
                log.error("消息扩展数据序列化失败，type: {}", messVo.getType(), e);
            }
        }
        
        return mess;
    }


    /**
     * 转换互动消息VO为实体
     * 支持完整的字段映射，包括content、commentId、replyId等扩展字段
     * 
     * 【关键修复】支持两种数据来源：
     * 1. 直接从MessageNotifyVO对象获取（instanceof判断）
     * 2. 从messageExt Map中提取（InteractionNotificationConsumeService封装的数据）
     * 
     * @param messVo 消息VO
     * @return 互动消息实体
     */
    private InteractionMessageEntity convertInteractionMessage(MessageBaseVO messVo) {
        log.info("【互动消息转换】开始转换，messVo类型: {}, messageExt类型: {}", 
                messVo.getClass().getSimpleName(), 
                messVo.getMessageExt() != null ? messVo.getMessageExt().getClass().getSimpleName() : "null");
        
        InteractionMessageEntity mess = new InteractionMessageEntity();
        mess.setSender(messVo.getSenderId());
        mess.setReceiver(messVo.getReceiverId());
        mess.setMessage(this.handleSpecialHtmlTag(HtmlUtil.xssEscape(messVo.getMessage())));
        mess.setSendDate(new Date());
        mess.setIsRead(false);
        mess.setImage(messVo.getImage());
        mess.setType(messVo.getType());
        
        // === 优先从messageExt Map中提取数据（最可靠的方式） ===
        // 经过测试发现：即使messVo是MessageNotifyVO类型，扩展字段也是存储在messageExt Map中
        // 所以优先判断messageExt是否为Map，如果是则从Map中提取
        if (messVo.getMessageExt() != null && messVo.getMessageExt() instanceof java.util.Map) {
            log.info("【互动消息转换】messageExt是Map类型，从Map中提取扩展字段");
            extractFieldsFromMessageExt(mess, (java.util.Map<String, Object>) messVo.getMessageExt());
        }
        // === 方式2：直接从MessageNotifyVO对象属性获取（备用方案） ===
        else if (messVo instanceof com.pai4j.domain.vo.request.mess.MessageNotifyVO) {
            log.info("【互动消息转换】识别为MessageNotifyVO类型，从对象属性提取扩展字段");
            com.pai4j.domain.vo.request.mess.MessageNotifyVO notifyVO = 
                (com.pai4j.domain.vo.request.mess.MessageNotifyVO) messVo;
            
            extractFieldsFromNotifyVO(mess, notifyVO);
        }
        else {
            log.warn("【互动消息转换】⚠️ 未能提取扩展字段！messVo类型: {}, messageExt: {}", 
                    messVo.getClass().getName(), messVo.getMessageExt());
        }
        
        log.info("【互动消息转换】转换完成，已设置字段 - messageType: {}, content: {}, commentId: {}, resourceId: {}, resourceType: {}, resourceTitle: {}", 
                mess.getMessageType(), mess.getContent(), mess.getCommentId(), 
                mess.getResourceId(), mess.getResourceType(), mess.getResourceTitle());
        
        return mess;
    }

    /**
     * 从MessageNotifyVO对象中提取扩展字段
     */
    private void extractFieldsFromNotifyVO(InteractionMessageEntity mess, 
            com.pai4j.domain.vo.request.mess.MessageNotifyVO notifyVO) {
        // 设置互动消息类型（具体的互动行为类型）
        mess.setMessageType(notifyVO.getMessageType());
        
        // 设置具体的互动内容
        if (StringUtils.isNotBlank(notifyVO.getContent())) {
            mess.setContent(this.handleSpecialHtmlTag(HtmlUtil.xssEscape(notifyVO.getContent())));
        }
        
        // 设置评论ID和回复ID
        mess.setCommentId(notifyVO.getCommentId());
        mess.setReplyId(notifyVO.getReplyId());
        
        // 设置被互动的资源信息
        mess.setResourceId(notifyVO.getResourceId());
        mess.setResourceType(notifyVO.getResourceType());
        if (StringUtils.isNotBlank(notifyVO.getResourceTitle())) {
            // 资源标题可能很长，截取前500个字符
            String resourceTitle = notifyVO.getResourceTitle();
            if (resourceTitle.length() > 500) {
                resourceTitle = resourceTitle.substring(0, 497) + "...";
            }
            mess.setResourceTitle(resourceTitle);
        }
        mess.setResourceImage(notifyVO.getResourceImage());
        mess.setResourceUrl(notifyVO.getResourceUrl());
        
        // 如果有扩展数据，序列化为JSON存储
        if (notifyVO.getMessageExt() != null) {
            try {
                mess.setExtData(JsonUtil.toJsonString(notifyVO.getMessageExt()));
            } catch (Exception e) {
                log.warn("互动消息扩展数据序列化失败: {}", e.getMessage());
            }
        }
    }

    /**
     * 从messageExt Map中提取扩展字段
     * （InteractionNotificationConsumeService将扩展字段封装到Map中）
     */
    private void extractFieldsFromMessageExt(InteractionMessageEntity mess, java.util.Map<String, Object> extData) {
        log.info("【互动消息】开始从messageExt提取扩展字段，Map keys: {}", extData.keySet());
        
        try {
            // 提取messageType
            if (extData.get("messageType") != null) {
                Integer messageType = toInteger(extData.get("messageType"));
                mess.setMessageType(messageType);
                log.info("【互动消息】设置messageType: {}", messageType);
            }
            
            // 提取content
            if (extData.get("content") != null) {
                String content = String.valueOf(extData.get("content"));
                if (StringUtils.isNotBlank(content) && !"null".equals(content)) {
                    mess.setContent(this.handleSpecialHtmlTag(HtmlUtil.xssEscape(content)));
                    log.info("【互动消息】设置content: {}", content);
                }
            }
            
            // 提取commentId
            if (extData.get("commentId") != null) {
                Long commentId = toLong(extData.get("commentId"));
                mess.setCommentId(commentId);
                log.info("【互动消息】设置commentId: {}", commentId);
            }
            
            // 提取replyId
            if (extData.get("replyId") != null) {
                Long replyId = toLong(extData.get("replyId"));
                mess.setReplyId(replyId);
                log.info("【互动消息】设置replyId: {}", replyId);
            }
            
            // 提取resourceId
            if (extData.get("resourceId") != null) {
                Long resourceId = toLong(extData.get("resourceId"));
                mess.setResourceId(resourceId);
                log.info("【互动消息】设置resourceId: {}", resourceId);
            }
            
            // 提取resourceType
            if (extData.get("resourceType") != null) {
                String resourceType = String.valueOf(extData.get("resourceType"));
                if (!"null".equals(resourceType)) {
                    mess.setResourceType(resourceType);
                    log.info("【互动消息】设置resourceType: {}", resourceType);
                }
            }
            
            // 提取resourceTitle
            if (extData.get("resourceTitle") != null) {
                String resourceTitle = String.valueOf(extData.get("resourceTitle"));
                if (StringUtils.isNotBlank(resourceTitle) && !"null".equals(resourceTitle)) {
                    if (resourceTitle.length() > 500) {
                        resourceTitle = resourceTitle.substring(0, 497) + "...";
                    }
                    mess.setResourceTitle(resourceTitle);
                    log.info("【互动消息】设置resourceTitle: {}", resourceTitle);
                }
            }
            
            // 提取resourceImage
            if (extData.get("resourceImage") != null) {
                String resourceImage = String.valueOf(extData.get("resourceImage"));
                if (!"null".equals(resourceImage)) {
                    mess.setResourceImage(resourceImage);
                    log.info("【互动消息】设置resourceImage: {}", resourceImage);
                }
            }
            
            // 提取resourceUrl
            if (extData.get("resourceUrl") != null) {
                String resourceUrl = String.valueOf(extData.get("resourceUrl"));
                if (!"null".equals(resourceUrl)) {
                    mess.setResourceUrl(resourceUrl);
                    log.info("【互动消息】设置resourceUrl: {}", resourceUrl);
                }
            }
            
            // 序列化整个extData到数据库的ext_data字段
            mess.setExtData(JsonUtil.toJsonString(extData));
            
            log.info("【互动消息】✅ 扩展字段提取完成: messageType={}, commentId={}, resourceId={}, resourceType={}", 
                    mess.getMessageType(), mess.getCommentId(), mess.getResourceId(), mess.getResourceType());
            
        } catch (Exception e) {
            log.error("【互动消息】❌ 从messageExt提取扩展字段失败", e);
        }
    }
    
    /**
     * 安全地将Object转换为Integer
     */
    private Integer toInteger(Object obj) {
        if (obj == null) return null;
        if (obj instanceof Integer) return (Integer) obj;
        if (obj instanceof Number) return ((Number) obj).intValue();
        try {
            return Integer.parseInt(String.valueOf(obj));
        } catch (Exception e) {
            log.warn("无法转换为Integer: {}", obj);
            return null;
        }
    }
    
    /**
     * 安全地将Object转换为Long
     */
    private Long toLong(Object obj) {
        if (obj == null) return null;
        if (obj instanceof Long) return (Long) obj;
        if (obj instanceof Number) return ((Number) obj).longValue();
        try {
            return Long.parseLong(String.valueOf(obj));
        } catch (Exception e) {
            log.warn("无法转换为Long: {}", obj);
            return null;
        }
    }



    /***
     * 处理特殊字符
     * @param content
     * @return
     */
    private String handleSpecialHtmlTag(String content) {
        if (StringUtils.isBlank(content)) {
            return content;
        }
        content = content.replaceAll("&lt;br&gt;", "<br>");
        content = content.replaceAll("&lt;b&gt;", "<b>");
        return content;
    }

    /**
     * 标记两个用户之间的消息为已读
     * 同时更新数据库的isRead字段和清除Redis未读计数缓存
     *
     * @param currUserId 当前用户ID（消息接收人）
     * @param chatUserId 聊天对象ID（消息发送人）
     * @return 标记已读的消息数量
     */
    public int markMessagesAsRead(String currUserId, String chatUserId) {
        log.info("标记消息为已读，接收人：{}，发送人：{}", currUserId, chatUserId);
        
        try {
            // 1. 更新数据库中的isRead字段
            int updatedCount = chatMessageDAO.markMessagesAsReadBySenderAndReceiver(currUserId, chatUserId);
            log.info("数据库更新完成，标记 {} 条消息为已读", updatedCount);
            
            // 2. 清除Redis未读计数缓存
            clearUnreadMessCountCache(currUserId, chatUserId);
            log.info("Redis未读计数缓存已清除");
            
            return updatedCount;
        } catch (Exception e) {
            log.error("标记消息已读失败，接收人：{}，发送人：{}，错误：{}", currUserId, chatUserId, e.getMessage(), e);
            throw new RuntimeException("标记消息已读失败", e);
        }
    }

    /**
     * 获取两个用户之间的未读消息数量
     *
     * @param currUserId 当前用户ID（消息接收人）
     * @param chatUserId 聊天对象ID（消息发送人）
     * @return 未读消息数量
     */
    public long getUnreadMessageCount(String currUserId, String chatUserId) {
        try {
            // 优先从数据库查询精确的未读消息数
            long count = chatMessageDAO.countUnreadMessages(currUserId, chatUserId);
            log.debug("查询未读消息数，接收人：{}，发送人：{}，未读数：{}", currUserId, chatUserId, count);
            return count;
        } catch (Exception e) {
            log.error("查询未读消息数失败，接收人：{}，发送人：{}，错误：{}", currUserId, chatUserId, e.getMessage(), e);
            return 0;
        }
    }

    /**
     * 生成聊天消息唯一标识
     * 【修复】使用消息内容的hash值而不是随机UUID，确保同一条消息的幂等性
     * 格式：{senderId}:{receiverId}:{messageHash}:{timestamp}
     *
     * @param messVo 消息VO
     * @return 消息唯一标识
     */
    private String generateMessageUuid(MessageBaseVO messVo) {
        String sessionIdOrUuid;
        if (StringUtils.isNotBlank(messVo.getSessionId())) {
            // 优先使用客户端提供的sessionId
            sessionIdOrUuid = messVo.getSessionId();
        } else {
            // 【关键修复】如果客户端没有提供sessionId，使用消息内容的hash值
            // 这样确保相同的消息内容+发送者+接收者+时间戳 = 相同的UUID
            // 即使消息队列重试，也能识别为同一条消息
            String messageContent = messVo.getMessage() != null ? messVo.getMessage() : "";
            Long timestamp = messVo.getTime() != null ? messVo.getTime() : System.currentTimeMillis();
            String hashSource = messVo.getSenderId() + ":" + messVo.getReceiverId() + ":" + messageContent + ":" + timestamp;
            sessionIdOrUuid = Integer.toHexString(hashSource.hashCode());
        }
        
        // 格式：senderId:receiverId:sessionId/hash:timestamp
        // 这样可以确保同一条消息的唯一性，即使重试也会被识别为同一条消息
        return String.format("%s:%s:%s:%s", 
                messVo.getSenderId(), 
                messVo.getReceiverId(), 
                sessionIdOrUuid,
                messVo.getTime() != null ? messVo.getTime() : System.currentTimeMillis());
    }
    
    /**
     * 生成互动消息唯一标识
     * 格式：{senderId}:{receiverId}:{type}:{bizId}:{timestamp}
     * 
     * 说明：
     * - bizId由commentId、replyId等业务ID组合生成
     * - 如果是MessageNotifyVO，则优先使用其业务ID
     * - 这样可以确保同一个互动行为（如同一条评论）不会产生重复的互动消息
     *
     * @param messVo 消息VO
     * @return 互动消息唯一标识
     */
    private String generateInteractionMessageUuid(MessageBaseVO messVo) {
        StringBuilder bizIdBuilder = new StringBuilder();
        
        // 如果是MessageNotifyVO，提取业务ID
        if (messVo instanceof com.pai4j.domain.vo.request.mess.MessageNotifyVO) {
            com.pai4j.domain.vo.request.mess.MessageNotifyVO notifyVO = 
                (com.pai4j.domain.vo.request.mess.MessageNotifyVO) messVo;
            
            // 1. 优先使用commentId和replyId构建业务ID（评论、回复场景）
            if (notifyVO.getCommentId() != null) {
                bizIdBuilder.append("c").append(notifyVO.getCommentId());
            }
            if (notifyVO.getReplyId() != null) {
                if (bizIdBuilder.length() > 0) {
                    bizIdBuilder.append("_");
                }
                bizIdBuilder.append("r").append(notifyVO.getReplyId());
            }
            
            // 2. 如果没有评论ID，使用resourceId + messageType（点赞、收藏等场景）
            // 这样可以保证同一个用户对同一个资源的同类型互动是唯一的
            if (bizIdBuilder.length() == 0 && notifyVO.getResourceId() != null) {
                bizIdBuilder.append("res").append(notifyVO.getResourceId());
                if (notifyVO.getMessageType() != null) {
                    bizIdBuilder.append("_mt").append(notifyVO.getMessageType());
                }
            }
        }
        
        // 3. 如果仍然没有业务ID，则使用sessionId或UUID（最后的fallback）
        if (bizIdBuilder.length() == 0) {
            if (StringUtils.isNotBlank(messVo.getSessionId())) {
                bizIdBuilder.append(messVo.getSessionId());
            } else {
                bizIdBuilder.append(java.util.UUID.randomUUID().toString().replace("-", ""));
            }
        }
        
        // UUID格式：senderId:receiverId:type:bizId:timestamp
        // 例如：
        // - 点赞：user123:user456:12:res789_mt1:1705737600000
        // - 评论：user123:user456:12:c789:1705737600000
        // - 回复：user123:user456:12:c789_r101:1705737600000
        return String.format("%s:%s:%s:%s:%s", 
                messVo.getSenderId(), 
                messVo.getReceiverId(), 
                messVo.getType(),
                bizIdBuilder.toString(),
                messVo.getTime() != null ? messVo.getTime() : System.currentTimeMillis());
    }
}
